package org.example.flinksql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;


public class Jdbc {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env, bsSettings);

        // create an input table
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_input (" +
                "id INT," +
                "username STRING," +
                "password STRING," +
                "nickname STRING," +
                "sex STRING," +
                "age INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_users'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");
        // create an output table
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_output (" +
                "id INT," +
                "username STRING," +
                "password STRING," +
                "nickname STRING," +
                "sex STRING," +
                "age INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_users_copy'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        // get table data with table api
        Table t_input = tbEnv.from("t_input").select($("*")).limit(5);
        // get table data with SQL
        Table t_output = tbEnv.sqlQuery("select * from t_output limit 5");

        t_input.execute().print();
        t_output.execute().print();

        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
